<?php

/**
 * @method static  \Redis getInstance($sign = '')
 */
class CoreQueueRedis implements CoreInterfaceAsync
{
    const SECTION = 'queue';
    const QUEUE_NAME = 'queue.core';

    /**
     * @param mixed ...$args
     * @return Redis
     */
    public static function instance(...$args): Redis
    {
        $redis = new \Redis();
        $redis->connect('127.0.0.1', '6379');
//        $redis->auth($password = 'hello');
        return $redis;
    }

    public static function consume()
    {
        $redis = self::getInstance();

        $i = 1;
        while (true) {
            try {
                $rs = $redis->blPop([static::QUEUE_NAME], 0);
                if ($rs) {
                    //文本数据落地

                    //解析队列内容
                    list($key, $value) = $rs;
                    $call = unserialize($value);
                    //执行具体方法
                    list($callback, $args) = $call;
                    call_user_func_array($callback, $args);
                    $i ++;
                }
            } catch (\Exception $e) {
                echo "error:", $e->getMessage(), PHP_EOL;
            }

        }
    }

    public static function produce(...$params)
    {
        var_dump("=======进入redis队列========");
        $redis = self::getInstance();
//        $params = [
//            '_q' => $callable,
//            '_params' => $params,
//            '_t' => microtime(true),
//            '_from' => gethostname(),
//        ];
        $redis->rPush(static::QUEUE_NAME, serialize($params));
    }
}